package com.atguigu.gmall.realtime.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.function.HbaseSink;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;

/**
 * @Author lzc
 * @Date 2023/7/29 13:58
 */
@Slf4j
public class HbaseUtil {
    
    /**
     * 获取 hbase 连接对象
     *
     * @return hbase 连接对象
     */
    public static Connection getHbaseConnection() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop162");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        Connection conn = null;
        try {
            conn = ConnectionFactory.createConnection(conf);
            System.out.println("conn = " + conn);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        return conn;
    }
    
    /**
     * 关闭 hbase 连接
     *
     * @param hbaseConn
     */
    public static void closeHbaseConnection(Connection hbaseConn) {
        if (hbaseConn != null) {
            try {
                hbaseConn.close();
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }
    }
    
    /**
     * 获取 hbase Admin 对象: 用来去建表, 删表,判断表是否存在
     *
     * @param hbaseConn
     * @return
     */
    public static Admin getHbaseAdmin(Connection hbaseConn) {
        try {
            return hbaseConn.getAdmin();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    /**
     * 关闭 admin 对象
     *
     * @param admin
     */
    public static void closeHbaseAdmin(Admin admin) {
        if (admin != null) {
            try {
                admin.close();
            } catch (IOException e) {
                e.printStackTrace();
                throw new RuntimeException(e);
            }
        }
    }
    
    /**
     * 根据传入的参数创建一个 hbase 表
     *
     * @param hbaseConn hbase 连接 对象
     * @param nameSpace 命名空间: 类型传统的数据的库名
     * @param table     表名
     * @param family    列族(只有一个列族)
     */
    public static void createHbaseTable(Connection hbaseConn,
                                        String nameSpace,
                                        String table,
                                        String family) {
        
        
        TableName tableName = TableName.valueOf(nameSpace, table);
        // 1. 判断表是否存在, 如果存在, 则直接返回
        try (Admin admin = hbaseConn.getAdmin()) { // 会自动关闭 admin 对象
            if (!admin.tableExists(tableName)) {
                System.out.println("admin = " + admin);
                // 2. 如果不存在, 则建表
                // family.getBytes(StandardCharsets.UTF_8) 不建议使用
                // 2.1 创建一个列族描述器
                ColumnFamilyDescriptor cfd = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family))
                    .build();
                // 2.2 表描述器
                TableDescriptor desc = TableDescriptorBuilder.newBuilder(tableName)
                    .setColumnFamily(cfd)
                    .build();
                // 2.3 创建表
                admin.createTable(desc);
                
                log.info("hbase的表: " + nameSpace + ":" + table + " 创建成功....");
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        
    }
    
    /**
     * 根据传入的条件删除指定的表
     *
     * @param hbaseConn hbase 连接 对象
     * @param nameSpace 命名空间: 类型传统的数据的库名
     * @param table     表名
     */
    public static void dropHbaseTable(Connection hbaseConn,
                                      String nameSpace,
                                      String table) {
        TableName tableName = TableName.valueOf(nameSpace, table);
        try (Admin admin = hbaseConn.getAdmin()) {
            if (admin.tableExists(tableName)) {
                // 1.先 disable
                admin.disableTable(tableName);
                // 2. drop
                admin.deleteTable(tableName);
                
                log.info("hbase的表: " + nameSpace + ":" + table + " 删除成功....");
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    public static SinkFunction<Tuple2<JSONObject, TableProcess>> getHbaseSink() {
        return new HbaseSink();
    }
    
    /**
     * 根据指定的参数, 向 hbase 写入一行数据
     *
     * @param hbaseConn    hbase 连接
     * @param nameSpaceStr 命名空间
     * @param tableNameStr 表名
     * @param rowKey       一行数据的 rowKey
     * @param columnFamily 列族
     * @param columns      要写入的列
     * @param data         列与对应的列值
     */
    public static void putOneRow(Connection hbaseConn,
                                 String nameSpaceStr,
                                 String tableNameStr,
                                 String rowKey,
                                 String columnFamily,
                                 String[] columns,
                                 JSONObject data) {
        // 1. 获取一张表
        TableName tableName = TableName.valueOf(nameSpaceStr, tableNameStr);
        try (Table table = hbaseConn.getTable(tableName)) {
            // 2. 创建一个 Put 对象
            Put put = new Put(Bytes.toBytes(rowKey));
            for (String column : columns) { //遍历所有的列
                String v = data.getString(column); // 获取改列值
                // v 有可能是 null, 如果是 null, 则Bytes.toBytes(v)会空指针
                if (v != null) {
                    put.addColumn(Bytes.toBytes(columnFamily),
                                  Bytes.toBytes(column),
                                  Bytes.toBytes(v)
                    );
                }
            }
            table.put(put);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        
    }
    
    public static void deleteOneRow(Connection hbaseConn,
                                    String nameSpaceStr,
                                    String tableNameStr,
                                    String rowKey) {
        TableName tableName = TableName.valueOf(nameSpaceStr, tableNameStr);
        try (Table table = hbaseConn.getTable(tableName)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    public static <T> T getOneRow(Connection hbaseConn,
                                  String nameSpace,
                                  String tableNameStr,
                                  String rowKey,
                                  Class<T> tClass) {
        
        // 2. 获取一个 Table 对象
        TableName tableName = TableName.valueOf(nameSpace, tableNameStr);
        try (Table table = hbaseConn.getTable(tableName)) {
            // 3. 查询指定的行
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = table.get(get);  //
            // 4. 遍历出所有的列
            List<Cell> cells = result.listCells();  // 每一列都是封装在一个 Cell 中
            //System.out.println(tableNameStr + " " + rowKey);
            
            T t = tClass.newInstance();
            for (Cell cell : cells) {  // {"列名": "值", "":"", '''}
                String k = Bytes.toString(CellUtil.cloneQualifier(cell));
                String v = Bytes.toString(CellUtil.cloneValue(cell));
                BeanUtils.setProperty(t, k, v);
            }
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    /**
     * 获取一个到 Hbase 的异步连接
     *
     * @return 到 Hbase 的异步连接对象
     */
    public static AsyncConnection getAsyncConnection() {
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "hadoop162");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        try {
            AsyncConnection asyncConnection = ConnectionFactory.createAsyncConnection(conf).get();
            System.out.println("conn = " + asyncConnection);
            return asyncConnection;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
        
    }
    
    /**
     * 关闭 hbase 的异步连接对象
     *
     * @param conn
     */
    public static void closeAsyncConnection(AsyncConnection conn) {
        
        try {
            if (conn != null) {
                conn.close();
            }
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
    
    /**
     * TODO
     * 根据参数从 hbase 中异步的查询一行数据, 然后把查询到数据封装 T 类型的对象中
     *
     * @param asyncHbaseConnection
     * @param nameSpace
     * @param tableNameStr
     * @param rowKey
     * @param tClass
     * @param <T>
     * @return
     */
    public static <T> T asyncGetOneRow(AsyncConnection asyncHbaseConnection,
                                       String nameSpace,
                                       String tableNameStr,
                                       String rowKey,
                                       Class<T> tClass) {
        TableName tableName = TableName.valueOf(nameSpace, tableNameStr);
        AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncHbaseConnection.getTable(tableName);
        Get get = new Get(Bytes.toBytes(rowKey));
        try {
            Result result = asyncTable.get(get).get();
            
            List<Cell> cells = result.listCells();
            T t = tClass.newInstance();
            for (Cell cell : cells) {
                String k = Bytes.toString(CellUtil.cloneQualifier(cell));
                String v = Bytes.toString(CellUtil.cloneValue(cell));
                BeanUtils.setProperty(t, k, v);
            }
            return t;
        } catch (Exception e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }
}
